package org.example.insert_overwrite;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class PaimonInsertOverwritePartition {
    public static void main(String[] args) {
        // 创建Flink流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().build();

        // INSERT OVERWRITE必须开启batch模式
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 创建Paimon catalog
        String createCatalogSql = "CREATE CATALOG paimon WITH (" +
                "    'type' = 'paimon'," +
                "    'warehouse' = 'file:///tmp/paimon'" +
                ");";
        System.out.println("正在执行创建Catalog的SQL：\n" + createCatalogSql);
        tableEnv.executeSql(createCatalogSql);

        String dropTableSql = "DROP TABLE IF EXISTS `paimon`.`default`.`user_behavior_copy`";
        System.out.println("正在执行删除表的SQL：\n" + dropTableSql);
        tableEnv.executeSql(dropTableSql);

        String createTableSql = "CREATE TABLE IF NOT EXISTS `paimon`.`default`.`user_behavior_copy` (\n" +
                "    user_id STRING,\n" +
                "    item_id STRING,\n" +
                "    behavior BIGINT,\n" +
                "    dt STRING,\n" +
                "    eventtime TIMESTAMP_LTZ(3),\n" +
                "    WATERMARK FOR eventtime AS eventtime - INTERVAL '5' SECOND,\n" +
                "    PRIMARY KEY (dt, user_id) NOT ENFORCED\n" +
                ") PARTITIONED BY (dt)\n" +
                ";";
        System.out.println("正在执行创建表的SQL：\n" + createTableSql);
        tableEnv.executeSql(createTableSql);

        String insertOverwrite1Sql = "INSERT OVERWRITE `paimon`.`default`.user_behavior_copy  partition(dt='2025-01-11') (user_id, item_id, behavior, eventtime)\n" +
                "SELECT user_id, item_id, behavior, cast(eventtime as TIMESTAMP_LTZ(3))\n" +
                "FROM `paimon`.`default`.`user_behavior`\n" +
                ";";
        System.out.println("正在执行插入覆盖分区2025-01-11的SQL：\n" + insertOverwrite1Sql);
        tableEnv.executeSql(insertOverwrite1Sql);

        String insertOverwrite2Sql = "INSERT OVERWRITE `paimon`.`default`.user_behavior_copy  partition(dt='2025-01-10') (user_id, item_id, behavior, eventtime)\n" +
                "SELECT user_id, item_id, behavior, cast(eventtime as TIMESTAMP_LTZ(3))\n" +
                "FROM `paimon`.`default`.`user_behavior`\n" +
                ";";
        System.out.println("正在执行插入覆盖分区2025-01-10的SQL：\n" + insertOverwrite2Sql);
        tableEnv.executeSql(insertOverwrite2Sql);
    }
}
